package com.example.rocketmq.demo.controller;

import com.example.rocketmq.demo.config.RocketMQConstant;
import com.example.rocketmq.demo.producer.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * RocketMQ 测试控制器
 */
@Slf4j
@RestController
@RequestMapping("/rocketmq")
public class RocketMQController {

    @Autowired
    private NormalProducer normalProducer;

    @Autowired
    private OrderlyProducer orderlyProducer;

    @Autowired
    private DelayProducer delayProducer;

    @Autowired
    private BatchProducer batchProducer;

    @Autowired
    private TransactionProducer transactionProducer;

    @Autowired
    private FilterProducer filterProducer;

    private AtomicLong orderIdGenerator = new AtomicLong(0);

    /**
     * 发送普通消息
     */
    @GetMapping("/normal")
    public String sendNormal(@RequestParam String message, @RequestParam String type) {
        try {
            switch (type) {
                case "sync":
                    SendResult syncResult = normalProducer.sendSync(message);
                    return "同步发送成功: " + syncResult.getMsgId();
                case "async":
                    normalProducer.sendAsync(message);
                    return "异步发送成功";
                case "oneway":
                    normalProducer.sendOneWay(message);
                    return "单向发送成功";
                default:
                    return "未知的发送类型";
            }
        } catch (Exception e) {
            log.error("发送普通消息失败", e);
            return "发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送顺序消息
     */
    @GetMapping("/orderly")
    public String sendOrderly(@RequestParam String message, @RequestParam(required = false) Long orderId) {
        try {
            if (orderId == null) {
                orderId = orderIdGenerator.incrementAndGet();
            }
            SendResult result = orderlyProducer.sendOrderly(message, orderId);
            return "顺序消息发送成功: " + result.getMsgId() + ", 订单ID: " + orderId;
        } catch (Exception e) {
            log.error("发送顺序消息失败", e);
            return "发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送延时消息
     */
    @GetMapping("/delay")
    public String sendDelay(@RequestParam String message, @RequestParam Integer delayLevel) {
        try {
            SendResult result = delayProducer.sendDelay(message, delayLevel);
            return "延时消息发送成功: " + result.getMsgId() + ", 延时等级: " + delayLevel;
        } catch (Exception e) {
            log.error("发送延时消息失败", e);
            return "发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送批量消息
     */
    @GetMapping("/batch")
    public String sendBatch(@RequestParam Integer count, @RequestParam String prefix) {
        try {
            List<String> messages = new ArrayList<>();
            for (int i = 0; i < count; i++) {
                messages.add(prefix + "-" + i);
            }
            SendResult result = batchProducer.sendBatch(messages);
            return "批量消息发送成功: " + result.getMsgId() + ", 消息数量: " + count;
        } catch (Exception e) {
            log.error("发送批量消息失败", e);
            return "发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送事务消息
     */
    @GetMapping("/transaction")
    public String sendTransaction(@RequestParam String message, @RequestParam String status) {
        try {
            TransactionSendResult result = transactionProducer.sendTransaction(message, status);
            return "事务消息发送成功: " + result.getMsgId() + ", 状态: " + status + ", 本地事务执行结果: " + result.getLocalTransactionState();
        } catch (Exception e) {
            log.error("发送事务消息失败", e);
            return "发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送标签过滤消息
     */
    @GetMapping("/filter/tag")
    public String sendFilterByTag(@RequestParam String message, @RequestParam String tag) {
        try {
            String actualTag;
            switch (tag) {
                case "A":
                    actualTag = RocketMQConstant.TAG_FILTER_A;
                    break;
                case "B":
                    actualTag = RocketMQConstant.TAG_FILTER_B;
                    break;
                case "C":
                    actualTag = RocketMQConstant.TAG_FILTER_C;
                    break;
                default:
                    return "无效的标签，请使用 A、B 或 C";
            }

            SendResult result = filterProducer.sendWithTag(message, actualTag);
            return "标签过滤消息发送成功: " + result.getMsgId() + ", 标签: " + actualTag;
        } catch (Exception e) {
            log.error("发送标签过滤消息失败", e);
            return "发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送SQL属性过滤消息
     */
    @GetMapping("/filter/sql")
    public String sendFilterBySQL(@RequestParam String message, @RequestParam Integer age) {
        try {
            SendResult result = filterProducer.sendWithProperty(message, RocketMQConstant.TAG_FILTER_A, "age", String.valueOf(age));
            return "SQL属性过滤消息发送成功: " + result.getMsgId() + ", 年龄: " + age;
        } catch (Exception e) {
            log.error("发送SQL属性过滤消息失败", e);
            return "发送失败: " + e.getMessage();
        }
    }
}
